package com.example.mq.mqserver;

import com.example.mq.common.*;
import com.example.mq.mqserver.core.BasicProperties;

import java.io.*;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class BrokerServer {
    private ServerSocket serverSocket=null;
    private VirtualHost virtualHost=new VirtualHost("default");
    private ConcurrentHashMap<String, Socket> sessions=new ConcurrentHashMap<>();
    private ExecutorService executorService=null;
    private volatile boolean runnable=true;

    public BrokerServer(int port) throws IOException {
        serverSocket=new ServerSocket(port);
    }

    public void start() throws IOException {
        System.out.println("[BrokerServer] 启动");
        executorService= Executors.newCachedThreadPool();
        try {
            while (runnable){
                Socket clientSocket=serverSocket.accept();
                executorService.submit(()->{
                   processConnection(clientSocket);
                });
            }
        }catch (SocketException e){
            System.out.println("[BrokerServer] 服务器停止运行！");
        }
    }

    public void stop() throws IOException {
        runnable=false;
        executorService.shutdownNow();
        serverSocket.close();
    }

    private void processConnection(Socket clientSocket) {
        try (InputStream inputStream=clientSocket.getInputStream();
             OutputStream outputStream=clientSocket.getOutputStream()){
            try (DataInputStream dataInputStream=new DataInputStream(inputStream);
                 DataOutputStream dataOutputStream=new DataOutputStream(outputStream)){
                while (true){
                    Request request=readRequest(dataInputStream);
                    Response response=process(request,clientSocket);
                    writeResponse(dataOutputStream,response);
                }
            }
        } catch (EOFException | SocketException e) {
            System.out.println("[BrokerServer] connection关闭！客户端地址为"+clientSocket.getInetAddress().toString()+":"+clientSocket.getPort());
        } catch (IOException | ClassNotFoundException | MqException e){
            System.out.println("[BrokerServer] connection出现异常！");
            e.printStackTrace();
        }finally {
            try {
                clientSocket.close();
                clearClosedSession(clientSocket);
            }catch (IOException e){
                e.printStackTrace();
            }
        }
    }

    private Request readRequest(DataInputStream dataInputStream) throws IOException {
        Request request=new Request();
        request.setType(dataInputStream.readInt());
        request.setLength(dataInputStream.readInt());
        byte[] payload=new byte[request.getLength()];
        int n=dataInputStream.read(payload);
        if(n!=request.getLength()){
            throw new IOException("请求格式出错");
        }
        request.setPayload(payload);
        return request;
    }

    private Response process(Request request, Socket clientSocket) throws IOException, ClassNotFoundException, MqException {
        BasicArguments basicArguments=(BasicArguments) BinaryTool.fromBytes(request.getPayload());
        System.out.println("[Request] rid="+basicArguments.getRid()+",channelId="+basicArguments.getChannelId()+",type="+request.getType()+",length="+request.getLength());
        boolean ok=true;
        if(request.getType()==0x1){
            sessions.put(basicArguments.getChannelId(),clientSocket);
            System.out.println("[BrokerServer] 创建channel完成！channelId="+basicArguments.getChannelId());
        }else if(request.getType()==0x2){
            sessions.remove(basicArguments.getChannelId());
            System.out.println("[BrokerServer] 删除channel完成！channelId="+basicArguments.getChannelId());
        }else if(request.getType()==0x3){
            ExchangeDeclareArguments arguments=(ExchangeDeclareArguments) basicArguments;
            ok=virtualHost.exchangeDeclare(arguments.getExchangeName(),arguments.getExchangeType(),arguments.isDurable(),arguments.isAutoDelete(),arguments.getArguments());
        }else if(request.getType()==0x4){
            ExchangeDeleteArguments arguments=(ExchangeDeleteArguments) basicArguments;
            ok=virtualHost.exchangeDelete(arguments.getExchangeName());
        }else if(request.getType()==0x5){
            QueueDeclareArguments arguments=(QueueDeclareArguments) basicArguments;
            ok=virtualHost.queueDeclare(arguments.getQueueName(),arguments.isDurable(),arguments.isExclusive(),arguments.isAutoDelete(),arguments.getArguments());
        }else if(request.getType()==0x6){
            QueueDeleteArguments arguments=(QueueDeleteArguments) basicArguments;
            ok=virtualHost.queueDelete(arguments.getQueueName());
        }else if(request.getType()==0x7){
            QueueBindArguments arguments=(QueueBindArguments) basicArguments;
            ok=virtualHost.queueBind(arguments.getQueueName(),arguments.getExchangeName(),arguments.getBindingKey());
        }else if(request.getType()==0x8){
            QueueUnbindArguments arguments=(QueueUnbindArguments) basicArguments;
            ok=virtualHost.queueUnbind(arguments.getQueueName(),arguments.getExchangeName());
        }else if(request.getType()==0x9){
            BasicPublishArguments arguments=(BasicPublishArguments) basicArguments;
            ok=virtualHost.basicPublish(arguments.getExchangeName(),arguments.getRoutingKey(),arguments.getBasicProperties(),arguments.getBody());
        }else if(request.getType()==0xa){
            BasicConsumeArguments arguments=(BasicConsumeArguments) basicArguments;
            ok=virtualHost.basicConsume(arguments.getConsumerTag(),arguments.getQueueName(),arguments.isAutoAck(),new Consumer(){
                @Override
                public void handleDelivery(String consumerTag, BasicProperties basicProperties, byte[] body) throws IOException,MqException{
                    Socket clientSocket=sessions.get(consumerTag);
                    if(clientSocket==null||clientSocket.isClosed()){
                        throw new MqException("[BrokerServer] 订阅消息的客户端已关闭");
                    }
                    SubScribeReturns subScribeReturns=new SubScribeReturns();
                    subScribeReturns.setChannelId(consumerTag);
                    subScribeReturns.setRid("");
                    subScribeReturns.setOk(true);
                    subScribeReturns.setConsumerTag(consumerTag);
                    subScribeReturns.setBasicProperties(basicProperties);
                    subScribeReturns.setBody(body);
                    byte[] payload=BinaryTool.toBytes(subScribeReturns);
                    Response response=new Response();
                    response.setType(0xc);
                    response.setLength(payload.length);
                    response.setPayload(payload);
                    DataOutputStream dataOutputStream=new DataOutputStream(clientSocket.getOutputStream());
                    writeResponse(dataOutputStream,response);
                }
            });
        }else if(request.getType()==0xb){
            BasicAckArguments arguments=(BasicAckArguments) basicArguments;
            ok=virtualHost.basicAck(arguments.getQueueName(),arguments.getMessageId());
        }else{
            throw new MqException("[BrokerServer] 位置类型，Type="+request.getType());
        }
        BasicReturns basicReturns=new BasicReturns();
        basicReturns.setChannelId(basicArguments.getChannelId());
        basicReturns.setRid(basicArguments.getRid());
        basicReturns.setOk(ok);
        byte[] payload=BinaryTool.toBytes(basicReturns);
        Response response=new Response();
        response.setType(response.getType());
        response.setLength(payload.length);
        response.setPayload(payload);
        System.out.println("[Request] rid="+basicReturns.getRid()+",channelId="+basicReturns.getChannelId()+",type="+response.getType()+",length="+response.getLength());
        return response;
    }

    private void writeResponse(DataOutputStream dataOutputStream, Response response) throws IOException {
        dataOutputStream.writeInt(response.getType());
        dataOutputStream.writeInt(response.getLength());
        dataOutputStream.write(response.getPayload());
        dataOutputStream.flush();
    }

    private void clearClosedSession(Socket clientSocket) {
        List<String> toDeleteChannelId=new ArrayList<>();
        for(Map.Entry<String,Socket> entry:sessions.entrySet()){
            if (entry.getValue()==clientSocket){
                toDeleteChannelId.add(entry.getKey());
            }
        }
        for (String channelId:toDeleteChannelId){
            sessions.remove(channelId);
        }
        System.out.println("[BrokerServer] 清除session完成，被处理的channelId="+toDeleteChannelId);
    }
}
